package com.example.demo.rabbitMQ.controller;

import com.example.demo.rabbitMQ.RabbitMqConnFactoryUtil;
import com.rabbitmq.client.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * <p>Description: </p>
 *
 * @author Eric Lee
 * @version v1.0.0
 * @since 2021/8/4 19:04
 **/
@Service
public class TransactionService {
    /**
     * 队列名称
     */
    private static final String TX_QUEUE = "tx_queue";

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    /**
     * 发送消息
     */
    public void publish()
            throws Exception {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost("/");
        factory.setHost(host);
        factory.setPort(port);
        Connection conn = factory.newConnection();
        // 创建信道
        Channel channel = conn.createChannel();
        // 声明队列
        channel.queueDeclare(TX_QUEUE, true, false, false, null);

        try {

            long startTime = System.currentTimeMillis();

            for (int i = 0; i < 10; i++) {
                // 声明事务
                channel.txSelect();
                String message = String.format("时间 => %s", System.currentTimeMillis());
                // 发送消息
                channel.basicPublish("", TX_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN,
                        message.getBytes("UTF-8"));
                // 提交事务
                channel.txCommit();
            }

            long endTime = System.currentTimeMillis();

            System.out.println("事务模式，发送10条数据，执行花费时间：" + (endTime - startTime) + "s");

        } catch (Exception e) {
            channel.txRollback();
        } finally {
            channel.close();
            conn.close();
        }
    }

    /**
     * 消费消息，单条消息消费
     *
     * @throws Exception
     */
    public void consume() throws Exception {

        Connection conn = RabbitMqConnFactoryUtil.getRabbitConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(TX_QUEUE, true, false, false, null);
        // 声明事务
        channel.txSelect();
        try {
            //单条消息获取进行消费
            GetResponse resp = channel.basicGet(TX_QUEUE, false);
            String message = new String(resp.getBody(), "UTF-8");
            System.out.println("收到消息：" + message);
            //消息拒绝
            // channel.basicReject(resp.getEnvelope().getDeliveryTag(), true);
            // 消息确认
            channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);
            // 提交事务
            channel.txCommit();
        } catch (Exception e) {
            // 回滚事务
            channel.txRollback();
        } finally {
            //关闭通道、连接
            channel.close();
            conn.close();
        }
    }
}
